package com.xian.interceptor;

import org.apache.kafka.clients.producer.ProducerInterceptor;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;

import java.util.Map;

/**
 * 统计发送成功或失败次数
 */
public class CountInterceptor implements ProducerInterceptor<String,String> {
    private int errorCounter = 0;
    private int successCounter = 0;

    //原样发送
    @Override
    public ProducerRecord<String, String> onSend(ProducerRecord<String, String> record) {
        return record;
    }

    @Override
    public void onAcknowledgement(RecordMetadata metadata, Exception exception) {
        //统计发送成功和失败次数，没有异常就是成功
        if(null == exception){
            successCounter++;
        } else {
            errorCounter++;
        }
        System.out.printf("当前发送成功次数为：%d, 发送失败次数为：%d\n", successCounter, errorCounter);


    }

    @Override
    public void close() {
        System.out.println("关闭count拦截器资源");
        System.out.printf("最终发送成功次数为：%d, 发送失败次数为：%d", successCounter, errorCounter);
    }

    @Override
    public void configure(Map<String, ?> configs) {

    }
}
